package com.zhu.function;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.zhu.utils.DIMUtils;
import com.zhu.utils.PhoenixUtil;
import com.zhu.utils.ZhuDruidDataSourceUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.SQLException;
import java.util.Locale;

public class DIMSinkFunction extends RichSinkFunction<JSONObject> {

    private DruidDataSource druidDataSource = null;

    /**
     * 初始化方法  获取数据化连接池
     * @param parameters
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        druidDataSource = ZhuDruidDataSourceUtil.createDataSource();
    }


    /**
     *每来一条数据执行一次
     * json sinktable 表名
     * @param value
     * @param context
     * @throws Exception
     */
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        //获取连接
        DruidPooledConnection connection = druidDataSource.getConnection();
        String sinkTable = value.getString("sinkTable");
        JSONObject data = value.getJSONObject("data");
        //写出数据
        /*
        {"sinkTable":"dim_base_trademark","database":"flink","xid":6115,"data":{"tm_name":"YSL","id":15},
        "commit":true,"type":"insert","table":"base_trademark","ts":1679742004}
         */
        //todo 获取数据类型 如果是更新 删除redis
        String type = value.getString("type");
        if("update".equals(type)){
            DIMUtils.delDimInfo(sinkTable.toUpperCase(Locale.ROOT),data.getString("id"));
        }

        PhoenixUtil.upsertValue(connection,sinkTable,data);  //数据写入Phoenix,如果写入失败


        //归还连接
        connection.close();
    }


}
